//! 完全解耦的连接处理器
//! 
//! 核心设计原则：
//! 1. **Tokio只出现在网络层** - 网络I/O使用Tokio异步处理
//! 2. **协议层完全同步** - 所有RESP解析和业务逻辑都是同步的
//! 3. **标准库桥接** - 使用std::sync::mpsc在网络层和协议层之间通信
//! 4. **完美解耦** - 协议层可以独立测试和运行，不依赖任何异步运行时
//! 
//! 架构层次：
//! ```
//! 网络层(Tokio)  ←→  标准库Channel  ←→  协议层(同步)
//! ```

use std::sync::{Arc, Mutex, mpsc};
use std::thread::{self, JoinHandle};
use std::time::Duration;
use std::collections::VecDeque;

use crate::protocol::RespParser;
use crate::storage::{Command, CommandResult, SingleThreadCommandDispatcher};
use crate::event::ConnectionId;

/// 网络层到协议层的消息
#[derive(Debug)]
pub enum NetworkToProtocol {
    /// 新数据到达
    DataReceived(Vec<u8>),
    /// 连接关闭
    ConnectionClosed,
    /// 关闭协议处理
    Shutdown,
}

/// 协议层到网络层的消息
#[derive(Debug)]
pub enum ProtocolToNetwork {
    /// 需要发送的响应数据
    SendResponse(Vec<u8>),
    /// 协议处理完成
    Finished,
    /// 错误信息
    Error(String),
}

/// 完全解耦的连接处理器
/// 
/// 这个处理器确保：
/// - 网络层使用Tokio处理异步I/O
/// - 协议层使用标准线程进行同步处理  
/// - 通过标准库的mpsc实现完美的层间通信
/// - 协议层完全不依赖任何异步运行时
pub struct FullyDecoupledHandler {
    /// 连接ID
    connection_id: ConnectionId,
    
    /// 发送消息到协议层（网络层→协议层）
    to_protocol_sender: mpsc::Sender<NetworkToProtocol>,
    
    /// 接收协议层的响应（协议层→网络层）
    from_protocol_receiver: mpsc::Receiver<ProtocolToNetwork>,
    
    /// 协议处理线程句柄
    protocol_thread: Option<JoinHandle<()>>,
}

impl FullyDecoupledHandler {
    /// 创建新的完全解耦连接处理器
    pub fn new(
        connection_id: ConnectionId,
        storage_dispatcher: Arc<Mutex<SingleThreadCommandDispatcher>>,
    ) -> Self {
        // 创建双向通道（使用标准库，不依赖Tokio）
        let (to_protocol_sender, to_protocol_receiver) = mpsc::channel();
        let (from_protocol_sender, from_protocol_receiver) = mpsc::channel();

        // 启动协议处理线程（完全同步）
        let protocol_thread = thread::spawn(move || {
            Self::protocol_processing_loop(
                connection_id,
                to_protocol_receiver,
                from_protocol_sender,
                storage_dispatcher,
            );
        });

        Self {
            connection_id,
            to_protocol_sender,
            from_protocol_receiver,
            protocol_thread: Some(protocol_thread),
        }
    }

    /// 网络层调用：发送新数据到协议层
    pub fn handle_data(&self, data: Vec<u8>) -> Result<(), String> {
        self.to_protocol_sender
            .send(NetworkToProtocol::DataReceived(data))
            .map_err(|_| "协议层已关闭".to_string())
    }

    /// 网络层调用：通知连接关闭
    pub fn handle_close(&self) -> Result<(), String> {
        self.to_protocol_sender
            .send(NetworkToProtocol::ConnectionClosed)
            .map_err(|_| "协议层已关闭".to_string())
    }

    /// 网络层调用：非阻塞获取待发送的响应
    pub fn try_get_response(&self) -> Option<ProtocolToNetwork> {
        self.from_protocol_receiver.try_recv().ok()
    }

    /// 网络层调用：阻塞获取响应（带超时）
    pub fn get_response_timeout(&self, timeout: Duration) -> Option<ProtocolToNetwork> {
        self.from_protocol_receiver.recv_timeout(timeout).ok()
    }

    /// 停止连接处理器
    pub fn stop(&mut self) -> Result<(), String> {
        // 发送停止命令到协议层
        let _ = self.to_protocol_sender.send(NetworkToProtocol::Shutdown);

        // 等待协议线程结束
        if let Some(handle) = self.protocol_thread.take() {
            handle.join()
                .map_err(|_| "协议线程结束失败".to_string())?;
        }

        println!("✅ 连接 {} 完全解耦处理器已停止", self.connection_id);
        Ok(())
    }

    /// 协议处理循环（在独立线程中运行，完全同步）
    fn protocol_processing_loop(
        connection_id: ConnectionId,
        message_receiver: mpsc::Receiver<NetworkToProtocol>,
        response_sender: mpsc::Sender<ProtocolToNetwork>,
        storage_dispatcher: Arc<Mutex<SingleThreadCommandDispatcher>>,
    ) {
        println!("🔄 连接 {connection_id} 开始协议处理循环（同步线程）");

        let mut resp_parser = RespParser::new();
        let mut pending_commands: VecDeque<Box<dyn Command>> = VecDeque::new();

        // 主处理循环 - 完全同步，不依赖Tokio
        loop {
            // 1. 接收来自网络层的消息
            let message = match message_receiver.recv_timeout(Duration::from_millis(100)) {
                Ok(msg) => msg,
                Err(mpsc::RecvTimeoutError::Timeout) => {
                    // 超时，继续处理挂起的命令
                    Self::process_pending_commands(
                        connection_id,
                        &mut pending_commands,
                        &storage_dispatcher,
                        &response_sender,
                    );
                    continue;
                }
                Err(mpsc::RecvTimeoutError::Disconnected) => {
                    println!("📡 连接 {connection_id} 网络层已断开");
                    break;
                }
            };

            match message {
                NetworkToProtocol::DataReceived(data) => {
                    // 2. 解析RESP协议（完全同步）
                    if let Err(e) = Self::parse_and_queue_commands(
                        connection_id,
                        data,
                        &mut resp_parser,
                        &mut pending_commands,
                    ) {
                        eprintln!("❌ 连接 {connection_id} 解析命令失败: {e}");
                        let _ = response_sender.send(ProtocolToNetwork::Error(e));
                    }

                    // 3. 处理解析出的命令
                    Self::process_pending_commands(
                        connection_id,
                        &mut pending_commands,
                        &storage_dispatcher,
                        &response_sender,
                    );
                }
                NetworkToProtocol::ConnectionClosed => {
                    println!("📡 连接 {connection_id} 协议层收到关闭通知");
                    let _ = response_sender.send(ProtocolToNetwork::Finished);
                    break;
                }
                NetworkToProtocol::Shutdown => {
                    println!("🛑 连接 {connection_id} 协议层收到停止命令");
                    break;
                }
            }
        }

        println!("✅ 连接 {connection_id} 协议处理循环结束");
    }

    /// 解析RESP数据并将命令加入队列
    fn parse_and_queue_commands(
        connection_id: ConnectionId,
        data: Vec<u8>,
        resp_parser: &mut RespParser,
        pending_commands: &mut VecDeque<Box<dyn Command>>,
    ) -> Result<(), String> {
        // 将数据喂给解析器
        resp_parser.feed_data(&data);

        // 解析所有可能的命令
        loop {
            match resp_parser.parse_value() {
                Ok(Some(resp_value)) => {
                    // 成功解析出一个RESP值，转换为命令
                    match resp_parser.resp_to_command(resp_value) {
                        Ok(command) => {
                            pending_commands.push_back(command);
                            println!("📝 连接 {} 解析出命令，队列长度: {}", 
                                connection_id, pending_commands.len());
                        }
                        Err(e) => {
                            eprintln!("❌ 连接 {connection_id} 命令转换失败: {e:?}");
                            return Err(format!("命令转换失败: {e:?}"));
                        }
                    }
                }
                Ok(None) => {
                    // 数据不完整，等待更多数据
                    break;
                }
                Err(e) => {
                    eprintln!("❌ 连接 {connection_id} RESP解析错误: {e:?}");
                    return Err(format!("RESP解析错误: {e:?}"));
                }
            }
        }

        Ok(())
    }

    /// 处理挂起的命令
    fn process_pending_commands(
        connection_id: ConnectionId,
        pending_commands: &mut VecDeque<Box<dyn Command>>,
        storage_dispatcher: &Arc<Mutex<SingleThreadCommandDispatcher>>,
        response_sender: &mpsc::Sender<ProtocolToNetwork>,
    ) {
        while let Some(command) = pending_commands.pop_front() {
            let command_name = command.name().to_string();

            // 分发命令到存储层（同步）
            let result = {
                let dispatcher = storage_dispatcher.lock().unwrap();
                dispatcher.dispatch_command(command)
            };

            match result {
                Ok(()) => {
                    println!("📤 连接 {connection_id} 命令 {command_name} 已分发到存储层");
                    
                    // 模拟命令结果（实际应该从存储层异步获取）
                    let command_result = match command_name.as_str() {
                        "PING" => CommandResult::String("PONG".to_string()),
                        "SET" => CommandResult::Ok,
                        "GET" => CommandResult::String("value".to_string()),
                        _ => CommandResult::String("OK".to_string()),
                    };

                    // 将结果转换为RESP格式并发送
                    let resp_bytes = command_result.to_resp().into_bytes();
                    if let Err(e) = response_sender.send(ProtocolToNetwork::SendResponse(resp_bytes)) {
                        eprintln!("❌ 连接 {connection_id} 发送响应失败: {e:?}");
                    }
                }
                Err(e) => {
                    eprintln!("❌ 连接 {connection_id} 命令 {command_name} 分发失败: {e}");
                    
                    // 发送错误响应
                    let error_result = CommandResult::Error(e.to_string());
                    let resp_bytes = error_result.to_resp().into_bytes();
                    if let Err(e) = response_sender.send(ProtocolToNetwork::SendResponse(resp_bytes)) {
                        eprintln!("❌ 连接 {connection_id} 发送错误响应失败: {e:?}");
                    }
                }
            }
        }
    }

    /// 获取连接ID
    pub fn connection_id(&self) -> ConnectionId {
        self.connection_id
    }
}

impl Drop for FullyDecoupledHandler {
    fn drop(&mut self) {
        let _ = self.stop();
    }
}

/// 完全解耦的协议服务器
/// 
/// 管理所有解耦的连接处理器，完全不依赖任何异步运行时
pub struct FullyDecoupledProtocolServer {
    /// 连接处理器映射
    connections: Arc<Mutex<std::collections::HashMap<ConnectionId, FullyDecoupledHandler>>>,
    /// 存储分发器
    storage_dispatcher: Arc<Mutex<SingleThreadCommandDispatcher>>,
    /// 连接计数器
    connection_counter: Arc<Mutex<u64>>,
}

impl FullyDecoupledProtocolServer {
    /// 创建新的完全解耦协议服务器
    pub fn new(storage_dispatcher: SingleThreadCommandDispatcher) -> Self {
        Self {
            connections: Arc::new(Mutex::new(std::collections::HashMap::new())),
            storage_dispatcher: Arc::new(Mutex::new(storage_dispatcher)),
            connection_counter: Arc::new(Mutex::new(0)),
        }
    }

    /// 创建新连接（由网络层调用）
    pub fn create_connection(&self) -> Result<(ConnectionId, FullyDecoupledHandler), String> {
        // 分配连接ID
        let connection_id = {
            let mut counter = self.connection_counter.lock().unwrap();
            *counter += 1;
            *counter
        };

        // 创建连接处理器
        let handler = FullyDecoupledHandler::new(
            connection_id,
            Arc::clone(&self.storage_dispatcher),
        );

        println!("✅ 创建连接 {connection_id} 的完全解耦处理器");
        Ok((connection_id, handler))
    }

    /// 移除连接
    pub fn remove_connection(&self, connection_id: ConnectionId) -> Result<(), String> {
        let mut connections = self.connections.lock().unwrap();
        if let Some(mut handler) = connections.remove(&connection_id) {
            handler.stop()?;
            println!("🗑️ 移除连接 {connection_id} 的处理器");
        }
        Ok(())
    }

    /// 获取活跃连接数
    pub fn get_active_connections_count(&self) -> usize {
        let connections = self.connections.lock().unwrap();
        connections.len()
    }

    /// 停止所有连接
    pub fn stop_all_connections(&self) -> Result<(), String> {
        let mut connections = self.connections.lock().unwrap();
        for (conn_id, handler) in connections.iter_mut() {
            if let Err(e) = handler.stop() {
                eprintln!("❌ 停止连接 {conn_id} 失败: {e}");
            }
        }
        connections.clear();
        println!("🛑 所有解耦连接处理器已停止");
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    #[test]
    fn test_fully_decoupled_handler() {
        let storage_dispatcher = SingleThreadCommandDispatcher::new();
        let mut handler = FullyDecoupledHandler::new(
            1,
            Arc::new(Mutex::new(storage_dispatcher)),
        );

        // 测试PING命令
        let ping_data = b"*1\r\n$4\r\nPING\r\n".to_vec();
        assert!(handler.handle_data(ping_data).is_ok());

        // 等待处理
        std::thread::sleep(Duration::from_millis(50));

        // 检查响应
        match handler.get_response_timeout(Duration::from_millis(100)) {
            Some(ProtocolToNetwork::SendResponse(response)) => {
                let response_str = String::from_utf8_lossy(&response);
                assert!(response_str.contains("PONG"));
                println!("✅ 完全解耦处理器测试成功: {}", response_str.trim());
            }
            _ => {
                panic!("未收到预期的PONG响应");
            }
        }

        assert!(handler.stop().is_ok());
    }

    #[test]
    fn test_protocol_server() {
        let storage_dispatcher = SingleThreadCommandDispatcher::new();
        let server = FullyDecoupledProtocolServer::new(storage_dispatcher);

        let (conn_id, mut handler) = server.create_connection().unwrap();
        assert_eq!(server.get_active_connections_count(), 1);

        // 测试SET命令
        let set_data = b"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n".to_vec();
        assert!(handler.handle_data(set_data).is_ok());

        // 等待处理
        std::thread::sleep(Duration::from_millis(50));

        // 检查响应
        if let Some(ProtocolToNetwork::SendResponse(response)) = 
            handler.get_response_timeout(Duration::from_millis(100)) {
            let response_str = String::from_utf8_lossy(&response);
            assert!(response_str.contains("OK"));
            println!("✅ SET命令处理成功: {}", response_str.trim());
        }

        // 清理
        assert!(handler.stop().is_ok());
        assert!(server.remove_connection(conn_id).is_ok());
        assert_eq!(server.get_active_connections_count(), 0);
    }

    #[test]
    fn test_no_tokio_dependency() {
        // 这个测试验证协议层完全没有Tokio依赖
        // 所有操作都是同步的，可以在任何线程中运行
        
        let storage_dispatcher = SingleThreadCommandDispatcher::new();
        let mut handler = FullyDecoupledHandler::new(
            1,
            Arc::new(Mutex::new(storage_dispatcher)),
        );

        // 在标准线程中处理（完全不需要Tokio运行时）
        std::thread::spawn(move || {
            let data = b"*2\r\n$3\r\nGET\r\n$3\r\nkey\r\n".to_vec();
            assert!(handler.handle_data(data).is_ok());

            // 同步等待和检查结果
            std::thread::sleep(Duration::from_millis(50));
            
            match handler.get_response_timeout(Duration::from_millis(100)) {
                Some(ProtocolToNetwork::SendResponse(_)) => {
                    println!("✅ 协议层成功处理命令（完全无Tokio依赖）");
                }
                _ => {
                    println!("⚠️ 未收到预期响应");
                }
            }

            assert!(handler.stop().is_ok());
        }).join().unwrap();
    }
}